package com.atguigu.flink.day07;

import com.atguigu.flink.beans.WaterSensor;
import com.atguigu.flink.func.WaterSensorMapFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author Felix
 * @date 2023/12/8
 * 侧输出流
 * 需求：对每个传感器，水位超过10的输出告警信息
 */
public class Flink05_side {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> wsDS = env
            .socketTextStream("hadoop102", 8888)
            .map(new WaterSensorMapFunction());

        OutputTag<String> warnTag = new OutputTag<String>("warnTag"){};
        SingleOutputStreamOperator<String> mainDS = wsDS.process(
            new ProcessFunction<WaterSensor, String>() {
                @Override
                public void processElement(WaterSensor ws, Context ctx, Collector<String> out) throws Exception {
                    Integer vc = ws.vc;
                    if (vc > 10) {
                        ctx.output(warnTag, ws.toString());
                    }else {
                        out.collect(ws.toString());
                    }
                }
            }
        );
        mainDS.print("主流：");
        mainDS.getSideOutput(warnTag).printToErr("侧流：");

        env.execute();
    }
}
